package org.lzfto.flink.demo.business.flinkTest.flinkObjects.connector.jdbc;

/**
 * Created by lzfto
 * Created time: 2025/05/25
 */

import java.io.Serializable;
import java.sql.Connection;
import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Enumeration;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.conversion.RowRowConverter;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
import org.lzfto.flink.demo.business.flinkTest.base.record.JdbcOptionRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CustomJdbcOutputOperation implements Serializable {

    private final  Logger LOGGER = LoggerFactory.getLogger(getClass());
    private static final long serialVersionUID = 1L;
	private final JdbcOptionRecord jdbcOption;

    public CustomJdbcOutputOperation(JdbcOptionRecord jdbcOption) {
        this.jdbcOption = jdbcOption;
    }

    public void executeInsert(RowData row, DataType dataType) throws SQLException {
    	//注意：只适合少量数据处理
        try (Connection conn = getConnection()) {
            String sql = buildInsertSQL(row, dataType);
            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                bindParameters(stmt, row,dataType);
                stmt.executeUpdate();
            }
        }
    }

    public void executeDelete(RowData row,DataType dataType) throws SQLException {
        try (Connection conn = getConnection()) {
            String sql = buildDeleteSQL(row);
            Row newRow=this.toRow(row,dataType);
            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                // 绑定主键参数（假设第一个字段是主键）
                if (row.isNullAt(0)) {
                    stmt.setObject(1, null);
                } else {
                    Object fieldValue = this.getFieldValue(newRow, 0);
                    stmt.setObject(1, fieldValue);
                }
                LOGGER.info("删除表{}的旧数据",jdbcOption.tableName());
                stmt.executeUpdate();
            }
        }
    }

    private String buildInsertSQL(RowData row, DataType dataType) {
        StringBuilder sql = new StringBuilder("INSERT INTO ")
            .append(this.jdbcOption.tableName())
            .append(" (");

        // 从dataType获取真实字段名
        List<String> fieldNameList=DataType.getFieldNames(dataType);
        int fieldCount = row.getArity();
        for (int i = 0; i < fieldCount; i++) {
            if (i > 0) {
                sql.append(",");
            }
            String fieldName=fieldNameList.get(i);                             
            sql.append(fieldName);
        }

        sql.append(") VALUES (");

        for (int i = 0; i < fieldCount; i++) {
            if (i > 0) {
                sql.append(",");
            }
            sql.append("?");
        }
        sql.append(")");

        return sql.toString();
    }

    private String buildDeleteSQL(RowData row) {
        return "DELETE FROM " + this.jdbcOption.tableName()
                + " WHERE " +this.jdbcOption.idColCode()+" = ?";
    }

    private void bindParameters(PreparedStatement stmt, RowData row, DataType dataType) throws SQLException {
        int fieldCount = row.getArity();
        Row newRow=this.toRow(row,dataType);
        for (int i = 0; i < fieldCount; i++) {
            int pos = i + 1; // JDBC参数从1开始
            if (row.isNullAt(i)) {
                stmt.setObject(pos, null);
            } else {
                Object fieldValue =this.getFieldValue(newRow, i);
                stmt.setObject(pos, fieldValue);
            }
        }
    }

    private Row toRow(RowData row, DataType dataType) {
        RowRowConverter converter = RowRowConverter.create(dataType);
        Row newRow=converter.toExternal(row);
        return newRow;
    }
    private Object getFieldValue(Row row, int pos) {
        return row.getField(pos);
    }


    private Connection getConnection() throws SQLException {
        //注意：通常此处代码无意义，可以删除。因为JDBC驱动4之后，SPI是自动加载的，不需要手动注册。
        //此处保留的原因是：因为已经写出来了
        if (!StringUtils.isNotEmpty(this.jdbcOption.driverClass())) {
            this.registerJdbcDriver(this.jdbcOption.driverClass());
        }

        return DriverManager.getConnection(
                this.jdbcOption.url(),this.jdbcOption.userName(), this.jdbcOption.password());
    }

    private void registerJdbcDriver(String driverClass) {
        try {
            if (!isDriverLoaded(driverClass)) {
                Class.forName(driverClass);
            }
        } catch (ClassNotFoundException e) {
            LOGGER.error("Driver class not found: {}", driverClass);
        }
    }
    private Boolean isDriverLoaded(String driverClass) {
        Enumeration<Driver> drivers = DriverManager.getDrivers();
        while (drivers.hasMoreElements()) {
            Driver driver = drivers.nextElement();
            if (driver.getClass().getName().equals(driverClass)) {
                return true;
            }
        }
        return false;
    }
}
